import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.common.util.SocketUtil;
import com.boarsoft.message.bean.Message;
import com.boarsoft.message.core.MessageService;

/**
 * 实现Socket服务端，接收并处理来自核心的短信报文<br>
 * 
 * 1、 短信输入适配器SmsInAdapter接受核心的SOCKET连接<br>
 * 2、 根据已知的短信模板编号，配置对应的队列。根据需要配置订阅人（微信网关）。<br>
 * 3、 额外配置一个sms_debit队列，配置其订阅人为短信银行。<br>
 * 4、 SmsInAdapter解析报文，报告中的每一行转换成一个Map，<br>
 * 根据其第一列“短信模板编号”（TEMPCODE）塞入对应的Message对象（不存在就创建）的content列表 <br>
 * 5、 创建单独的Message，设置其目标队列为sms_debit，content为原始报文。<br>
 * 6、 短信输出适配器SmsOutAdapter将来自sms_debit队列的消息的content通过SOCKET形式发送给短信网关<br>
 * 
 * 
 * @author Mac_J
 *
 */
public class SmsInAdapter implements Runnable, CompletionHandler<AsynchronousSocketChannel, Object> {
	private static final Logger log = LoggerFactory.getLogger(SmsInAdapter.class);
	private String splitor = "|*|";
	private final String batchFileTempCode = "9999";

	protected Map<String, String> templateMata;// 模板
	protected Map<String, String> msgRecipients;// 消息分发
	protected MessageService localMsgSvc1;
	protected SmsOutAdapter smsOutAdapter;
	protected BatchForFile batchForFile;

	/** */
	protected ExecutorService ioThreadPool;
	/** */
	protected ExecutorService threadPool;
	/** */
	// protected ScheduledExecutorService scheduler;
	/** */
	protected AsynchronousChannelGroup asyncChannelGroup;
	/** */
	protected AsynchronousServerSocketChannel asyncServerSocketChannel;

	protected Charset charset = Charset.forName("GBK");
	// protected CharsetDecoder charsetDecoder = charset.newDecoder();

	@PostConstruct
	public void init() throws IOException {
		this.asyncChannelGroup = AsynchronousChannelGroup.withThreadPool(this.ioThreadPool);
		// this.asyncServerSocketChannel =
		// AsynchronousServerSocketChannel.open(this.asyncChannelGroup)
		// .bind(new InetSocketAddress("20.3.5.176", 9000));
		this.asyncServerSocketChannel = AsynchronousServerSocketChannel.open(this.asyncChannelGroup)
				.bind(new InetSocketAddress(9000));
		// this.scheduler.scheduleWithFixedDelay(this,
		// RpcConfig.HEARTBEAT_PERIOD, RpcConfig.HEARTBEAT_PERIOD,
		// TimeUnit.MILLISECONDS);
		// this.status = STATUS_RUNNING;
		// this.ioThreadPool.execute(this);
		log.info("SmsInAdapter initiated successfully.");
		this.run();
	}

	public String process(String text) {
		log.info("Received message {}", text);
		// 将Socket收到的报文转换成Message对象
		Message m = convert(text);
		if (m == null) {
			log.error("Error on create message from text = {}", text);
			return "Invalid message text";
		}
		try {
			localMsgSvc1.put(m);
		} catch (Exception e) {
			log.error("Error on put message", e);
			return e.getMessage();
		}
		// 批量文件处理,暂时不区分微信、短信开头的标志
		doBatchFile(m);
		// return "0接收成功";//修改返回值 2018年3月常规版本
		return "0";
	}

	public Message convert(String text) {
		String[] fields = null;
		String templateCode = null;
		List<Map<String, String>> list = new ArrayList<>();
		Map<String, String> msgMap = new HashMap<String, String>();
		String msg = msdBuild(text);
		String[] cols = Util.splitByWord(msg, splitor);

		if (cols != null && cols.length > 0) {
			templateCode = cols[6];
		}
		String smsTag = msgTag().get(templateCode);
		if (smsTag != null && smsTag.trim() != "") {
			fields = Util.splitByWord(smsTag, splitor);
		}
		if (fields != null && cols != null /* && fields.length == cols.length */) {
			for (int i = 0; i < fields.length && i < cols.length; i++) {
				msgMap.put(fields[i], cols[i].trim());
			}
		}
		list.add(msgMap);
		Message m = new Message();
		m.setTitle("sms");
		String tempCode = msgMap.get("TEMPCODE");
		if (batchFileTempCode.equals(tempCode)) {
			m.setTarget("sms_debit_batch");
		} else {
			m.setTarget("sms_debit");
		}
		m.setAttachment(removeMsgRecipients(text));
		// m.setContent(msgMap);
		m.setContent(list);
		m.addRecipients(getRecipientList(msgMap));
		// m.addRecipient("smsOutAdapter");
		// m.addRecipient("wechat");
		m.setId(RandomUtil.genUUID());
		m.setFrom("core");
		return m;
		// String attachment = msg;
		// // 0101 动账
		// String target = msgMap.get("SENDACCOUNT");// 发送账号
		// String showAccount = msgMap.get("SIGNACCOUNT");// 显示尾号
		// String tranDate = msgMap.get("TRANDATE");// 交易日期
		// String tranTime = msgMap.get("TRANTIME");// 交易时间
		// String balance = msgMap.get("BALANCE");// 余额
		// String tranmt = msgMap.get("TRANAMT");// 交易金额
		// String tranSign = msgMap.get("TRANSIGN");// 借贷标志 1--借方 2--贷方
		// String summary = msgMap.get("SUMMARY");// 摘要
		// String aimAcountNO = msgMap.get("AIMACCOUNTNO");// 对方账户
		// String aimUserName = msgMap.get("AIMUSERNAME");// 对方户名
		//
		// //0102 销户
		// String pinType = msgMap.get("PINTYPE");//介质类型
	}

	@Override
	public void run() {
		asyncServerSocketChannel.accept(null, this);
	}

	@Override
	public void completed(AsynchronousSocketChannel asc, Object attachement) {
		final ByteBuffer buf = ByteBuffer.allocateDirect(2048);
		StringBuilder sb = new StringBuilder();
		try {
			// buf.clear();
			while (asc.read(buf).get(3000, TimeUnit.MILLISECONDS) > 0) {
				buf.flip();
				CharBuffer cb = Charset.forName("GBK").decode(buf);
				String s = sb.append(cb).toString();
				// F5心跳
				if ("00000000".equals(s)) {
					asc.write(ByteBuffer.wrap("f5HeartBeatOK".getBytes("GBK")));
					return;
				}
				s = Util.replace(s, "<EOF>", "");
				s = Util.replace(s, "\r", "");
				s = Util.replace(s, "\n", "");
				String[] rows = Util.splitByWord(s, "\n");
				if (rows != null && rows.length > 0) {
					String resultVal = rows[0].trim();
					String reVal = this.process(resultVal);
					asc.write(ByteBuffer.wrap(reVal.getBytes("GBK")));
					return;
				}
				buf.clear();
			}
			buf.clear();
		} catch (Exception e) {
			log.error("Error on process socket request", e);
		} finally {
			// if ("00000000".equals(sb.toString())) {
			// if (log.isDebugEnabled()) {
			// log.debug("SmsInAdapter receive message:{}", sb.toString());
			// }
			// } else {
			// log.info("SmsInAdapter receive message:{}", sb.toString());
			// }
			SocketUtil.closeChannel(asc);
			this.asyncServerSocketChannel.accept(null, this);
		}
	}

	@Override
	public void failed(Throwable e, Object attachment) {
		log.error("Failed to process socket request", e);
	}

	private Map<String, String> msgTag() {
		/*
		 * Map<String, String> map = new HashMap<String, String>();
		 * map.put("0101",
		 * "TRANSCODE|*|CHANNELNO|*|TRANDATE|*|TRANTIME|*|TRANORG|*|TEMPCODE|*|TASKID|*|SIGNACCOUNT|*|MSGNUMBER|*|SEQUENCE|*|BALANCE|*|TRANAMT|*|CURRENCYTYPE|*|TRANSIGN|*|TRANTYPE|*|SUMMARY|*|AIMACCOUNTNO|*|AIMUSERNAME|*|CIFNO|*|OPTYPE|*|ADDRESS|*|TIPFLAG|*|SENDACCOUNT"
		 * ); // map.put("0102", //
		 * "TRANSCODE|*|CHANNELNO|*|TRANDATE|*|TRANTIME|*|TRANORG|*|TEMPCODE|*|SIGNACCOUNT|*|CURRENCYTYPE|*|TRANAMT|*|PINTYPE|*|ADDRESS|*|SENDACCOUNT"
		 * ); return map;
		 */
		return getTemplateMata();
	}

	private String msdBuild(String text) {
		String msg = null;
		if (text != null && text.trim() != "") {
			msg = Util.replace(text, "<EOF>", "");
			msg = Util.replace(msg, "\r", "");
			msg = Util.replace(msg, "\n", "");
			msg = msg.trim();
		}
		return msg;
	}

	/**
	 * 传给微信、短信的报文，去掉转发内容，转发内容为第一位置
	 * 
	 * @param originalMsg
	 * @return
	 */
	private String removeMsgRecipients(String originalMsg) {
		String reVal = null;
		if (StringUtils.isNotEmpty(originalMsg) && originalMsg.contains(splitor)) {
			int firstP = originalMsg.indexOf(splitor);
			reVal = originalMsg.substring(firstP + splitor.length(), originalMsg.length());
		}
		return reVal + "\r\n<EOF>";
	}

	/**
	 * 根据报文转发数值 获取消息接收者
	 * 
	 * @param msgMap
	 * @return
	 */
	private List<String> getRecipientList(Map<String, String> msgMap) {
		List<String> msgRecipientsForWardList = null;
		String msgRecipientsVal = msgMap.get("MSGRECIPIENTS");// 消息分发数值
																// eg短信：10000000
		if (StringUtils.isNotEmpty(msgRecipientsVal)) {
			String msgRecipientsForWard = this.getMsgRecipients().get(msgRecipientsVal);// 消息分发
																						// eg短信且微信：smsOutAdapter,wechat
			if (StringUtils.isNotEmpty(msgRecipientsForWard)) {
				String[] msgRecipientsForWardArr = msgRecipientsForWard.split(",");
				msgRecipientsForWardList = Arrays.asList(msgRecipientsForWardArr);
			}
		}
		return msgRecipientsForWardList;
	}

	/**
	 * 处理批量短信
	 * 
	 * @param m
	 * @return
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public void doBatchFile(Message m) {
		if (m == null || m.getContent() == null) {
			log.error("批量文件处理失败，结果:{}", "Message 为空");
			return;
		}
		List<Map> list = (List) m.getContent();
		if (list == null || list.size() <= 0) {
			log.error("批量文件处理失败，结果:{}", "Message——m.getContent() 为空");
			return;
		}
		Map<String, String> map = list.get(0);
		if (map == null) {
			log.error("批量文件处理失败，结果:{}", "Message——m.getContent()——map 为空");
			return;
		}
		String tempCode = map.get("TEMPCODE");
		if (!batchFileTempCode.equals(tempCode)) {// 9999批量文件报文
			// log.error("批量文件处理失败，结果:{}","批量文件报文格式错误"); 实时报文不输出
			return;
		}
		String fileName = map.get("FILENAME");
		log.info("处理批量文件开始：{}", fileName);
		try {
			batchForFile.batchFileSMS(fileName);
		} catch (Throwable e) {
			log.error("批量文件处理失败，文件：{}", fileName);
		}
		// log.info("处理批量文件结束：{}",fileName); 异步处理
	}

	public BatchForFile getBatchForFile() {
		return batchForFile;
	}

	public void setBatchForFile(BatchForFile batchForFile) {
		this.batchForFile = batchForFile;
	}

	public ExecutorService getIoThreadPool() {
		return ioThreadPool;
	}

	public void setIoThreadPool(ExecutorService ioThreadPool) {
		this.ioThreadPool = ioThreadPool;
	}

	public ExecutorService getThreadPool() {
		return threadPool;
	}

	public void setThreadPool(ExecutorService threadPool) {
		this.threadPool = threadPool;
	}

	public MessageService getLocalMsgSvc1() {
		return localMsgSvc1;
	}

	public void setLocalMsgSvc1(MessageService localMsgSvc1) {
		this.localMsgSvc1 = localMsgSvc1;
	}

	public SmsOutAdapter getSmsOutAdapter() {
		return smsOutAdapter;
	}

	public void setSmsOutAdapter(SmsOutAdapter smsOutAdapter) {
		this.smsOutAdapter = smsOutAdapter;
	}

	public Map<String, String> getTemplateMata() {
		return templateMata;
	}

	public void setTemplateMata(Map<String, String> templateMata) {
		this.templateMata = templateMata;
	}

	public Map<String, String> getMsgRecipients() {
		return msgRecipients;
	}

	public void setMsgRecipients(Map<String, String> msgRecipients) {
		this.msgRecipients = msgRecipients;
	}

}
